package com.raylu.day04;

import com.raylu.utils.IntSource;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;

// 排序历史数据
public class Example3HistoryDataSort {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .keyBy(r -> 1)
                .process(new MyProcess())
                .print();

        env.execute();
    }

    public static class MyProcess extends KeyedProcessFunction<Integer, Integer, String> {
        // 列表状态变量用来保存历史数据
        private ListState<Integer> history;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            history = getRuntimeContext().getListState(
                    new ListStateDescriptor<Integer>("history", Types.INT)
            );
        }

        @Override
        public void processElement(Integer value, Context ctx, Collector<String> out) throws Exception {
            // 将到达的数据添加到列表状态变量中
            history.add(value);

            // 对历史数据进行排序
            // 将数据从列表状态变量中取出，然后添加到ArrayList中
            ArrayList<Integer> integers = new ArrayList<>();
            for (Integer i : history.get()) {
                integers.add(i);
            }

            // 排序
            integers.sort(new Comparator<Integer>() {
                @Override
                public int compare(Integer i1, Integer i2) {
                    return i1 - i2;
                }
            });

            StringBuilder result = new StringBuilder();
            for (int i = 0; i < integers.size() - 1; i++) {
                result.append(integers.get(i) + " => ");
            }
            result.append(integers.get(integers.size() - 1));
            out.collect(result.toString());
        }
    }
}
